Flink: Refactor to use the BaseEqualityDeltaWriter. by openinx · Pull Request #4264 · apache/iceberg

openinx · 2022-03-04T02:44:37Z

This PR is trying to refactor the flink write path to remove those underlying old writers to the partition specified writers ( Introduced from @aokolnychyi ).

I was reviewing the PR #4132 from @dungdm93, which is the similar thing that this one is trying to accomplish. I tried to rewrite it because I wanted to try out the details of the design for myself and get a better understanding of those new writers. In this way, I think I can provide more background & details design from my perspective. After high level design are okay, then I think we can make the whole PR to be smaller ones and let's collaborate together to get this work done @dungdm93 .

dungdm93 · 2022-03-04T03:31:00Z

core/src/main/java/org/apache/iceberg/deletes/EqualityDeleteWriter.java

  private final int[] equalityFieldIds;
  private final SortOrder sortOrder;
  private DeleteFile deleteFile = null;
+  private long rowOffset = 0;


How about name recordCount or rowCount?
IMO rowOffset is specific for tracking record write to file, recordCount is more general.

I see the BaseFile is using the recordCount, I think it's better to keep this consistent with the BaseFile's recordCount :-)

openinx · 2022-03-08T04:25:46Z

FYI @rdblue , @aokolnychyi , @dungdm93 . I think this draft PR is ready to review now. Mind to take a look when you have a chance ?

openinx · 2022-03-08T04:35:03Z

core/src/main/java/org/apache/iceberg/io/PartitioningWriter.java

   * @param partition a partition or null if the spec is unpartitioned
   */
-  void write(T row, PartitionSpec spec, StructLike partition);
+  PathOffset write(T row, PartitionSpec spec, StructLike partition);


As the PartitioningWriter will have multiple partition writers in its internal implementation, so we can not just introduce the two methods in this interface:

/** * Returns the file path that are currently opened. * * @return the current file path. */ CharSequence location(); /** * Returns the row offset that are currently writing, starting from 0. * * @return the current row offset. */ long rowOffset();

Because those location() and rowOffset() are binded to specific writers. In theory, different writers should have different location() and rowOffset(). So here we add a return value PathOffset here to get the latest wrote path & offset for the wrote row.

emkornfield · 2022-03-10T05:40:26Z

core/src/main/java/org/apache/iceberg/io/BaseEqualityDeltaWriter.java

+import org.apache.iceberg.util.StructLikeMap;
+import org.apache.iceberg.util.StructProjection;
+
+public class BaseEqualityDeltaWriter<T> implements EqualityDeltaWriter<T> {


Are there any style guidelines for documentation? Typically its nice to have javadoc on al public classes/members that aren't overwritten.

emkornfield · 2022-03-10T05:40:59Z

core/src/main/java/org/apache/iceberg/io/BaseEqualityDeltaWriter.java

+
+public class BaseEqualityDeltaWriter<T> implements EqualityDeltaWriter<T> {
+
+  private final ThreadLocal<PositionDelete<T>> posDelete = ThreadLocal.withInitial(PositionDelete::create);


maybe document why thread local is used here?

emkornfield · 2022-03-10T05:45:25Z

core/src/main/java/org/apache/iceberg/io/FileWriter.java

  void write(T row);

+  /**
+   * Returns the file path that are currently opened.


Suggested change

* Returns the file path that are currently opened.

* Returns the file path that is currently opened.

Is "opened" important here? what happens if the writer is closed?

emkornfield · 2022-03-10T05:49:36Z

core/src/main/java/org/apache/iceberg/io/FileWriter.java

+  CharSequence location();
+
+  /**
+   * Returns the row offset that are currently writing, starting from 0.


Suggested change

* Returns the row offset that are currently writing, starting from 0.

* Returns the row offset that will be written to next, starting from 0.

Is this interface intended to be seekable at any point? If not maybe call this rowsWritten which might be clearer?

As we've discussed in https://github.com/apache/iceberg/pull/4264/files#r819238913, I agree recordCount is a good & consistent name.

openinx · 2022-03-24T03:43:14Z

core/src/main/java/org/apache/iceberg/io/BaseEqualityDeltaWriter.java

+    if (!retireOldKey(asStructLike.apply(key), spec, partition)) {
+      equalityWriter.write(key, spec, partition);
+    }


I think this will be needed to reconsider because of this PR: #4364

openinx · 2022-04-02T08:03:39Z

CC @rdblue @aokolnychyi , Would you mind to take a look when you have a chance ?

github-actions · 2024-08-07T00:14:08Z

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.

github-actions · 2024-08-15T00:13:42Z

This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.

openinx added this to the Iceberg 0.14.0 Release milestone Mar 4, 2022

github-actions bot added core flink labels Mar 4, 2022

openinx marked this pull request as draft March 4, 2022 02:45

dungdm93 reviewed Mar 4, 2022

View reviewed changes

openinx marked this pull request as ready for review March 8, 2022 04:26

openinx commented Mar 8, 2022

View reviewed changes

openinx requested review from aokolnychyi and rdblue March 8, 2022 08:28

openinx mentioned this pull request Mar 9, 2022

Flink: Add disk-based insertedRowMap to resolve the OOM while ingesting. #4298

Closed

emkornfield reviewed Mar 10, 2022

View reviewed changes

openinx mentioned this pull request Mar 21, 2022

Flink - Fix incorrect / old row being written into delta files when using upsert mode #4364

Merged

openinx commented Mar 24, 2022

View reviewed changes

openinx force-pushed the flink-writer-refactor branch from 588e7b3 to 6f329e1 Compare March 28, 2022 09:57

github-actions bot added the data label Mar 28, 2022

openinx force-pushed the flink-writer-refactor branch from 3bc727c to be391f7 Compare March 29, 2022 09:51

openinx added 9 commits April 18, 2022 15:51

Flink: Refactor to use the BaseEqualityDeltaWriter.

68fd511

tmp

18eca26

Replace the TaskWriter.

10daaa5

Temporary fix.

87eaaaf

Remove the old delta writer.

c98365c

Fix the broken unit tests.

76979b8

Fix the broken unit tests.

4335865

Address the unit tests.

1ecc3f0

Address the unit tests.

d9dc0a9

openinx added 9 commits April 18, 2022 15:51

Refactor the rowOffset to recordCount.

521e72d

Address comments.

25071c4

Add unit tests.

232c4fb

Remove the orc special cases.

3b71206

Minor fixes.

8aef88d

Fix the checkstyles.

ed45f5d

Improve unit tests and address comments.

e9657fc

Fix the broken TestFlinkUpsert.

5793eeb

Rebase to the latest master & Address the unit tests.

86f83fa

openinx force-pushed the flink-writer-refactor branch from be391f7 to 86f83fa Compare April 18, 2022 07:52

rdblue removed this from the Iceberg 0.14.0 Release milestone Jun 28, 2022

github-actions bot added the stale label Aug 7, 2024

github-actions bot closed this Aug 15, 2024


		public class BaseEqualityDeltaWriter<T> implements EqualityDeltaWriter<T> {

		private final ThreadLocal<PositionDelete<T>> posDelete = ThreadLocal.withInitial(PositionDelete::create);

	* Returns the file path that are currently opened.
	* Returns the file path that is currently opened.

	* Returns the row offset that are currently writing, starting from 0.
	* Returns the row offset that will be written to next, starting from 0.

Conversation

openinx commented Mar 4, 2022

Uh oh!

dungdm93 Mar 4, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

openinx commented Mar 8, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

openinx commented Apr 2, 2022

Uh oh!

github-actions bot commented Aug 7, 2024

Uh oh!

github-actions bot commented Aug 15, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

dungdm93 Mar 4, 2022 •

edited

Loading